package car_test;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

public class mr_isshangyong {



    public static class MyMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();

       /* Map函数处理输入的每行文本数据
 *
         * @param key 输入的键，通常不用于处理逻辑
 * @param value 输入的值，代表文件中的一行文本
 * @param context 上下文对象，用于输出中间结果到Reduce
 * @throws IOException 当读写数据发生错误时
 * @throws InterruptedException 当任务被中断时
 */
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            // 将输入的文本按GBK编码转换为字符串，以处理中文字符
            String line = new String(value.getBytes(), 0, value.getLength(), "GBK");
            // 使用正则表达式分割字符串，同时考虑逗号和引号的情况
            String[] strs = line.trim().split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
            // 检查分割后的数组是否非空且长度为39，以确保数据完整性
            if (strs != null && strs.length == 39) {
                // 获取数组中的第11个元素作为分类依据
                String kind = strs[10];
                // 判断分类，如果不是'non-profitable'，则将其归类为'profitable'
                if (!kind.equals("non-profitable")) {
                    kind = "profitable";
                }
                // 设置word为当前分类
                word.set(kind);
                // 输出分类到Reduce阶段，计数为1
                context.write(word, one);
            }
        }
    }

    public static class MyReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
        private Map<Text, Long> map = new HashMap<Text, Long>();
        private long total = 0;

        /**
         * Reduce任务处理函数
         * 本函数重写自父类，用于处理reduce任务中的键值对数据
         * 主要功能是汇总每个键对应的值的总和，并更新全局总计数器
         * 同时，将每个键及其对应的值的总和存储在映射中，以便后续处理
         *
         * @param key 当前处理的键，代表了数据的分类标识
         * @param values 与当前键相关联的值的集合，代表了需要汇总的数据
         * @param context 上下文对象，用于与Hadoop框架进行交互，传递中间数据或状态信息
         * @throws IOException 当输入/输出操作发生错误时抛出此异常
         * @throws InterruptedException 当线程被中断时抛出此异常
         */
        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            // 初始化当前键对应值的总和
            long sum = 0;
            // 遍历值的集合，计算总和
            for (IntWritable val : values) {
                sum += val.get();
            }
            // 更新全局总计数器
            total += sum;
            // 将当前键及其值的总和存储在映射中
            map.put(new Text(key), sum);
        }

        /**
         * 执行清理操作，将处理后的数据写出
         * 此方法通常用于MapReduce框架的Reducer阶段，用于在处理所有输入后执行清理操作
         * 它遍历之前收集的所有键值对，计算每个键对应值的平均数，并将结果写出
         *
         * @param context 上下文对象，用于与MapReduce框架交互，通过它来访问输入、输出以及报告进度等
         * @throws IOException 如果在写入过程中发生I/O错误
         * @throws InterruptedException 如果在执行过程中被中断
         */
        public void cleanup(Context context) throws IOException, InterruptedException {
            // 获取map中所有键的集合，这里的键是Text类型，代表不同的数据标识
            Set<Text> ks = map.keySet();
            // 遍历每个键，计算其对应值除以总数的结果，并将结果写出
            for (Text t : ks) {
                // 创建一个DoubleWritable对象，用于存储计算结果
                DoubleWritable result = new DoubleWritable((double) map.get(t) / total);
                // 将键和计算结果组合成键值对，并通过context写出
                context.write(new Text(t.toString() + " vehicles"), result);
            }
        }

    }
    /**
     * 程序的入口点，用于执行MapReduce作业
     * @param args 命令行参数
     * @throws Exception 如果作业执行失败，抛出异常
     */
    public static void main(String args[]) throws Exception{

        // 设置NameNode的IP地址
        String namenode_ip = "192.168.128.130";
        // 构建HDFS的URL
        String hdfs = "hdfs://" + namenode_ip + ":8020";
        // 创建并配置Hadoop的Configuration对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", hdfs);

        // 设置作业的名称
        String jobName = "v1";
        // 创建Job实例
        Job job = Job.getInstance(conf, jobName);
        // 设置Map任务的类
        job.setMapperClass(MyMapper.class);
        // 设置Map任务输出的键类型
        job.setMapOutputKeyClass(Text.class);
        // 设置Map任务输出的值类型
        job.setMapOutputValueClass(IntWritable.class);
        // 设置Reduce任务的类
        job.setReducerClass(MyReducer.class);
        // 设置作业最终输出的键类型
        job.setOutputKeyClass(Text.class);
        // 设置作业最终输出的值类型
        job.setOutputValueClass(DoubleWritable.class);

        // 设置输入文件的路径
        String dataDir = "/car/data.csv";
        // 设置输出目录的路径
        String outputDir = "/Output/v1";
        // 创建输入路径的Path对象
        Path inPath = new Path(hdfs + dataDir);
        // 创建输出路径的Path对象
        Path outPath = new Path(hdfs + outputDir);
        // 为作业添加输入路径
        FileInputFormat.addInputPath(job, inPath);
        // 设置作业的输出路径
        FileOutputFormat.setOutputPath(job, outPath);

        // 获取FileSystem实例
        FileSystem fs = FileSystem.get(conf);
        // 如果输出路径已存在，则删除它
        if (fs.exists(outPath)) {
            fs.delete(outPath, true);
        }

        // 打印作业运行信息
        System.out.println("Job: " + jobName + " is running...");
        // 等待作业完成
        job.waitForCompletion(true);
        // 根据作业完成情况打印结果
        if (job.waitForCompletion(true)) {
            System.out.println("success!");
        } else {
            System.out.println("failed!");
            System.exit(1);
        }
    }


}
